package com.atguigu.gmall.realtime.app.func;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.beans.TableProcess;
import org.apache.flink.api.common.state.BroadcastState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ReadOnlyBroadcastState;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.util.Collector;

import java.sql.*;
import java.util.*;

/**
 * @author Felix
 * @date 2023/3/15
 */
public class BaseDbTableProcessFunction extends BroadcastProcessFunction<JSONObject,String,JSONObject> {

    private MapStateDescriptor<String, TableProcess> mapStateDescriptor;

    private Map<String,TableProcess> mapConfig = new HashMap<>();

    public BaseDbTableProcessFunction(MapStateDescriptor<String, TableProcess> mapStateDescriptor) {
        this.mapStateDescriptor = mapStateDescriptor;
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        Connection conn = DriverManager.getConnection("jdbc:mysql://hadoop202:3306/gmall0926_config?" +
            "user=root&password=123456&useUnicode=true&" +
            "characterEncoding=utf8&serverTimeZone=Asia/Shanghai&useSSL=false");

        String sql = "select * from gmall0926_config.table_process where sink_type = 'dwd'";
        PreparedStatement ps = conn.prepareStatement(sql);
        ResultSet rs = ps.executeQuery();
        ResultSetMetaData metaData = rs.getMetaData();
        while (rs.next()){
            JSONObject jsonObj = new JSONObject();
            for (int i = 1; i <= metaData.getColumnCount(); i++) {
                String columnName = metaData.getColumnName(i);
                Object columnValue = rs.getObject(i);
                jsonObj.put(columnName,columnValue);
            }
            TableProcess tableProcess = jsonObj.toJavaObject(TableProcess.class);
            String sourceTable = tableProcess.getSourceTable();
            String sourceType = tableProcess.getSourceType();
            String key = sourceTable +":" + sourceType;
            mapConfig.put(key,tableProcess);
        }

        rs.close();
        ps.close();
        conn.close();
    }

    //处理主流业务数据
    @Override
    public void processElement(JSONObject jsonObj, ReadOnlyContext ctx, Collector<JSONObject> out) throws Exception {
        //{"database":"gmall0926","xid":458,"data":{"tm_name":"三星1","logo_url":"/static/default.jpg","id":1},"old":{"tm_name":"三星"},"commit":true,"type":"update","table":"base_trademark","ts":1678498329}
        //获取广播广播
        ReadOnlyBroadcastState<String, TableProcess> broadcastState = ctx.getBroadcastState(mapStateDescriptor);
        //获取操作的业务数据库表名
        String table = jsonObj.getString("table");
        //获取操作类型
        String type = jsonObj.getString("type");
        //拼接key
        String key = table + ":" + type;
        //根据key到广播状态中获取对应的配置信息
        TableProcess tableProcess = null;

        if((tableProcess = broadcastState.get(key)) != null ||
            (tableProcess = mapConfig.get(key)) != null ){
            JSONObject dataJsonObj = jsonObj.getJSONObject("data");

            //过滤掉不需要传递的属性
            String sinkColumns = tableProcess.getSinkColumns();
            filterColumn(dataJsonObj,sinkColumns);
            //在向下游传递数据前，补充输出的目的地(kafka的主题)
            String sinkTable = tableProcess.getSinkTable();
            dataJsonObj.put("sink_table",sinkTable);
            //将需要分流的事实数据向下游传递
            out.collect(dataJsonObj);
        }
    }

    //过滤掉不需要传递的属性
    private void filterColumn(JSONObject dataJsonObj, String sinkColumns) {
        String[] columnArr = sinkColumns.split(",");
        List<String> columnList = Arrays.asList(columnArr);
        Set<Map.Entry<String, Object>> entrySet = dataJsonObj.entrySet();
        entrySet.removeIf(entry->!columnList.contains(entry.getKey()));
    }

    //"op":"r" : {"before":null,"after":{"source_table":"activity_info","source_type":"ALL","sink_table":"dim_activity_info","sink_type":"dim","sink_columns":"id,activity_name,activity_type,activity_desc,start_time,end_time,create_time","sink_pk":"id","sink_extend":null},"source":{"version":"1.6.4.Final","connector":"mysql","name":"mysql_binlog_source","ts_ms":0,"snapshot":"false","db":"gmall0926_config","sequence":null,"table":"table_process","server_id":0,"gtid":null,"file":"","pos":0,"row":0,"thread":null,"query":null},"op":"r","ts_ms":1678429765000,"transaction":null}
    //"op":"c" : {"before":null,"after":{"source_table":"a","source_type":"ALL","sink_table":"dim_a","sink_type":"dim","sink_columns":"id,name","sink_pk":"id","sink_extend":null},"source":{"version":"1.6.4.Final","connector":"mysql","name":"mysql_binlog_source","ts_ms":1678429887000,"snapshot":"false","db":"gmall0926_config","sequence":null,"table":"table_process","server_id":1,"gtid":null,"file":"mysql-bin.000011","pos":773930,"row":0,"thread":null,"query":null},"op":"c","ts_ms":1678429887821,"transaction":null}
    //"op":"u" : {"before":{"source_table":"a","source_type":"ALL","sink_table":"dim_a","sink_type":"dim","sink_columns":"id,name","sink_pk":"id","sink_extend":null},"after":{"source_table":"a","source_type":"ALL","sink_table":"dim_a","sink_type":"dim","sink_columns":"id,name,age","sink_pk":"id","sink_extend":null},"source":{"version":"1.6.4.Final","connector":"mysql","name":"mysql_binlog_source","ts_ms":1678429924000,"snapshot":"false","db":"gmall0926_config","sequence":null,"table":"table_process","server_id":1,"gtid":null,"file":"mysql-bin.000011","pos":774267,"row":0,"thread":null,"query":null},"op":"u","ts_ms":1678429924215,"transaction":null}
    //"op":"d" : {"before":{"source_table":"a","source_type":"ALL","sink_table":"dim_a","sink_type":"dim","sink_columns":"id,name,age","sink_pk":"id","sink_extend":null},"after":null,"source":{"version":"1.6.4.Final","connector":"mysql","name":"mysql_binlog_source","ts_ms":1678429966000,"snapshot":"false","db":"gmall0926_config","sequence":null,"table":"table_process","server_id":1,"gtid":null,"file":"mysql-bin.000011","pos":774643,"row":0,"thread":null,"query":null},"op":"d","ts_ms":1678429966145,"transaction":null}
    //处理广播流配置数据
    @Override
    public void processBroadcastElement(String jsonStr, Context ctx, Collector<JSONObject> out) throws Exception {
        //为了处理方便，将jsonStr转换为jsonObj
        JSONObject jsonObj = JSON.parseObject(jsonStr);
        //获取对配置表操作类型
        String op = jsonObj.getString("op");
        //获取广播状态
        BroadcastState<String, TableProcess> broadcastState = ctx.getBroadcastState(mapStateDescriptor);
        if("d".equals(op)){
            //从配置表中删除了一条数据
            TableProcess before = jsonObj.getObject("before", TableProcess.class);
            //判断是否删除的是事实配置
            String sinkType = before.getSinkType();
            if("dwd".equals(sinkType)){
                //将对应事实数据配置信息从广播状态中删除掉
                String sourceTable = before.getSourceTable();
                String sourceType = before.getSourceType();
                String key = sourceTable +":" + sourceType;
                broadcastState.remove(key);
                mapConfig.remove(key);
            }
        }else{
            //对配置表做的是读取、添加或者修改
            TableProcess after = jsonObj.getObject("after", TableProcess.class);
            //判断是否操作的是事实配置
            String sinkType = after.getSinkType();
            if("dwd".equals(sinkType)){
                //将最新的配置信息放到广播状态中
                String sourceTable = after.getSourceTable();
                String sourceType = after.getSourceType();
                String key = sourceTable +":" + sourceType;
                broadcastState.put(key,after);
                mapConfig.put(key,after);
            }
        }
    }
}
